package org.apache.flink.connector.scylladb.table;

import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.connector.scylladb.config.ScyllaDBConfigOptions;
import org.apache.flink.connector.scylladb.config.ScyllaDBConnectorConstants;
import org.apache.flink.table.connector.format.DecodingFormat;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.factories.FactoryUtil;
import org.apache.flink.table.types.DataType;

import java.util.HashSet;
import java.util.Set;

import static org.apache.flink.table.factories.FactoryUtil.createTableFactoryHelper;

/**
 * ScyllaDB 动态表源工厂，用于创建 ScyllaDB 表源
 */
public class ScyllaDBDynamicTableSourceFactory implements DynamicTableSourceFactory {

    @Override
    public String factoryIdentifier() {
        return ScyllaDBConnectorConstants.IDENTIFIER;
    }

    @Override
    public Set<ConfigOption<?>> requiredOptions() {
        final Set<ConfigOption<?>> options = new HashSet<>();
        options.add(ScyllaDBConfigOptions.URL);
        options.add(ScyllaDBConfigOptions.KEY_SPACE);
        options.add(ScyllaDBConfigOptions.TABLE_NAME);
        return options;
    }

    @Override
    public Set<ConfigOption<?>> optionalOptions() {
        final Set<ConfigOption<?>> options = new HashSet<>();
        options.add(ScyllaDBConfigOptions.USERNAME);
        options.add(ScyllaDBConfigOptions.PASSWORD);
        options.add(ScyllaDBConfigOptions.SCAN_FETCH_SIZE);
        options.add(ScyllaDBConfigOptions.SCAN_PARTITION_COLUMN);
        options.add(ScyllaDBConfigOptions.SCAN_PARALLELISM);
        options.add(ScyllaDBConfigOptions.LOOKUP_CACHE_MAX_ROWS);
        options.add(ScyllaDBConfigOptions.LOOKUP_CACHE_TTL);
        options.add(ScyllaDBConfigOptions.LOOKUP_MAX_RETRIES);
        options.add(ScyllaDBConfigOptions.LOOKUP_AGGREGATE_MODE);
        options.add(ScyllaDBConfigOptions.LOOKUP_AGGREGATE_FIELD_NAME);
        options.add(ScyllaDBConfigOptions.CONNECTION_POOL_LOCAL_SIZE);
        options.add(ScyllaDBConfigOptions.CONNECTION_POOL_REMOTE_SIZE);
        options.add(ScyllaDBConfigOptions.CONNECTION_TIMEOUT);
        options.add(ScyllaDBConfigOptions.REQUEST_TIMEOUT);
        options.add(ScyllaDBConfigOptions.CONSISTENCY_LEVEL);
        options.add(ScyllaDBConfigOptions.SSL_ENABLED);
        options.add(ScyllaDBConfigOptions.SSL_KEYSTORE_PATH);
        options.add(ScyllaDBConfigOptions.SSL_KEYSTORE_PASSWORD);
        options.add(ScyllaDBConfigOptions.SSL_TRUSTSTORE_PATH);
        options.add(ScyllaDBConfigOptions.SSL_TRUSTSTORE_PASSWORD);
        return options;
    }

    @Override
    public DynamicTableSource createDynamicTableSource(Context context) {
        // 使用 FactoryUtil 帮助器验证选项并获取配置
        final FactoryUtil.TableFactoryHelper helper = createTableFactoryHelper(this, context);
        
        // 验证所有选项
        helper.validate();
        
        // 获取配置
        final ReadableConfig config = helper.getOptions();
        
        // 获取物理数据类型
        final DataType physicalDataType = context.getCatalogTable().getResolvedSchema().toPhysicalRowDataType();
        
        // 创建并返回 ScyllaDB 表源
        return new ScyllaDBDynamicTableSource(
                config.get(ScyllaDBConfigOptions.URL),
                config.get(ScyllaDBConfigOptions.KEY_SPACE),
                config.get(ScyllaDBConfigOptions.TABLE_NAME),
                config.get(ScyllaDBConfigOptions.USERNAME),
                config.get(ScyllaDBConfigOptions.PASSWORD),
                config.get(ScyllaDBConfigOptions.SCAN_FETCH_SIZE),
                config.getOptional(ScyllaDBConfigOptions.SCAN_PARTITION_COLUMN).orElse(null),
                config.get(ScyllaDBConfigOptions.SCAN_PARALLELISM),
                config.get(ScyllaDBConfigOptions.LOOKUP_CACHE_MAX_ROWS),
                config.get(ScyllaDBConfigOptions.LOOKUP_CACHE_TTL),
                config.get(ScyllaDBConfigOptions.LOOKUP_MAX_RETRIES),
                config.get(ScyllaDBConfigOptions.CONNECTION_POOL_LOCAL_SIZE),
                config.get(ScyllaDBConfigOptions.CONNECTION_POOL_REMOTE_SIZE),
                config.get(ScyllaDBConfigOptions.CONNECTION_TIMEOUT),
                config.get(ScyllaDBConfigOptions.REQUEST_TIMEOUT),
                config.get(ScyllaDBConfigOptions.CONSISTENCY_LEVEL),
                config.get(ScyllaDBConfigOptions.SSL_ENABLED),
                config.getOptional(ScyllaDBConfigOptions.SSL_KEYSTORE_PATH).orElse(null),
                config.getOptional(ScyllaDBConfigOptions.SSL_KEYSTORE_PASSWORD).orElse(null),
                config.getOptional(ScyllaDBConfigOptions.SSL_TRUSTSTORE_PATH).orElse(null),
                config.getOptional(ScyllaDBConfigOptions.SSL_TRUSTSTORE_PASSWORD).orElse(null),
                config.get(ScyllaDBConfigOptions.LOOKUP_AGGREGATE_MODE),
                config.get(ScyllaDBConfigOptions.LOOKUP_AGGREGATE_FIELD_NAME),
                physicalDataType,
                context.getCatalogTable().getResolvedSchema()
        );
    }
} 